// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

// This example showcases various ways to work with Datasets. It's
// intended to be paired with the documentation.

#include <nebula/core/api.h>
#include <nebula/compute/api.h>
#include <nebula/csv/api.h>
#include <nebula/acero/exec_plan.h>
#include <nebula/compute/expression.h>

#include <nebula/dataset/dataset.h>
#include <nebula/dataset/plan.h>
#include <nebula/dataset/scanner.h>

#include <nebula/io/interfaces.h>
#include <nebula/io/memory.h>
#include <nebula/io/stdio.h>

#include <nebula/filesystem/filesystem.h>


#include <turbo/utility/status.h>

#include <nebula/util/vector.h>

#include <iostream>
#include <vector>

namespace ds = nebula::dataset;
namespace cp = nebula::compute;

char kLeftRelationCsvData[] = R"csv(lkey,shared,ldistinct
1,4,7
2,5,8
11,20,21
3,6,9)csv";

char kRightRelationCsvData[] = R"csv(rkey,shared,rdistinct
1,10,13
124,10,11
2,11,14
5,12,15)csv";

turbo::Result<std::shared_ptr<nebula::dataset::Dataset>> create_dataSet_from_csv_data(
        bool is_left) {
    const nebula::io::IOContext &io_context = nebula::io::default_io_context();
    std::shared_ptr<nebula::io::InputStream> input;
    std::string csv_data = is_left ? kLeftRelationCsvData : kRightRelationCsvData;
    std::cout << csv_data << std::endl;
    std::string_view sv = csv_data;
    input = nebula::io::BufferReader::from_string(std::string(sv));
    auto read_options = nebula::csv::ReadOptions::defaults();
    auto parse_options = nebula::csv::ParseOptions::defaults();
    auto convert_options = nebula::csv::ConvertOptions::defaults();

    // Instantiate TableReader from input stream and options
    TURBO_MOVE_OR_RAISE(std::shared_ptr<nebula::csv::TableReader> table_reader,
                        nebula::csv::TableReader::create(io_context, input, read_options,
                                                         parse_options, convert_options));

    // read table from CSV file
    TURBO_MOVE_OR_RAISE(auto maybe_table, table_reader->read());
    auto ds = std::make_shared<nebula::dataset::InMemoryDataset>(maybe_table);
    turbo::Result<std::shared_ptr<nebula::dataset::InMemoryDataset>> result(std::move(ds));
    return result;
}

turbo::Status do_hash_join() {
    nebula::dataset::internal::Initialize();

    TURBO_MOVE_OR_RAISE(auto l_dataset, create_dataSet_from_csv_data(true));
    TURBO_MOVE_OR_RAISE(auto r_dataset, create_dataSet_from_csv_data(false));

    auto l_options = std::make_shared<nebula::dataset::ScanOptions>();
    // create empty projection: "default" projection where each field is mapped to a
    // field_ref
    l_options->projection = cp::project({}, {});

    auto r_options = std::make_shared<nebula::dataset::ScanOptions>();
    // create empty projection: "default" projection where each field is mapped to a
    // field_ref
    r_options->projection = cp::project({}, {});

    // construct the scan node
    auto l_scan_node_options = nebula::dataset::ScanNodeOptions{l_dataset, l_options};
    auto r_scan_node_options = nebula::dataset::ScanNodeOptions{r_dataset, r_options};

    nebula::acero::Declaration left{"scan", std::move(l_scan_node_options)};
    nebula::acero::Declaration right{"scan", std::move(r_scan_node_options)};

    nebula::acero::HashJoinNodeOptions join_opts{nebula::acero::JoinType::INNER,
            /*in_left_keys=*/{"lkey"},
            /*in_right_keys=*/{"rkey"},
            /*filter*/ nebula::compute::literal(true),
            /*output_suffix_for_left*/ "_l",
            /*output_suffix_for_right*/ "_r"};

    nebula::acero::Declaration hashjoin{
            "hashjoin", {std::move(left), std::move(right)}, join_opts};

    // expected columns l_a, l_b
    TURBO_MOVE_OR_RAISE(std::shared_ptr<nebula::Table> response_table,
                        nebula::acero::declaration_to_table(std::move(hashjoin)));

    std::cout << "Results :"<<std::endl<< response_table->to_string() << std::endl;

    return turbo::OkStatus();
}

int main(int argc, char **argv) {
    auto status = do_hash_join();
    if (!status.ok()) {
        std::cerr << "Error occurred: " << status.message() << std::endl;
        return EXIT_FAILURE;
    }
    return EXIT_SUCCESS;
}
